package com.pp.web.map;

import rx.Observable;
import rx.Scheduler;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.schedulers.Schedulers;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;

public class ObservbleTest {


    public static void main(String [] args) throws Exception{
        test();

    }


    //用指定的名称新建一个线程
    public static Scheduler getNamedScheduler(final String name) {
        System.out.println("创建一个新线程");
        return Schedulers.from(Executors.newCachedThreadPool(new ThreadFactory() {

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r,name);
            }
        }));
    }

    //打印当前线程的名称
    public static void threadInfo(String caller) {
        System.out.println(caller + " => " + Thread.currentThread().getName());
    }

    public static void test() throws ExecutionException, InterruptedException {

        Observable defer = Observable.defer(new Func0() {
            private String test(){
                System.out.println("1");
                return "1";
            }
            @Override
            public Observable call() {
                return Observable.just(test());
            }
        }).subscribeOn(getNamedScheduler("new thread"));


        Future future = defer.toBlocking().toFuture();
        Object o = future.get();


    }

    public static void test2() throws Exception{

        Observable defer = Observable.defer(new Func0() {
            private String test(){
                System.out.println("111111");
                return "1";
            }
            @Override
            public Observable call() {
                return Observable.just(test());
            }
        });


        Future<Integer> future = defer.doOnTerminate(new Action0() {
            @Override
            public void call() {
                System.out.println("订阅即将被终止.....................");
            }
        }).doOnUnsubscribe(new Action0() {
            @Override
            public void call() {
                System.out.println("观察者取消订阅了它生成的Observable.....................");
            }
        }).doOnCompleted(new Action0() {
            @Override
            public void call() {
                System.out.println("完成.....................");
            }
        }).doOnError(new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                System.out.println("error!");
            }
        }).toBlocking().toFuture();

        future.get();
    }

}
